Redis集群下过期key监听

您所在的位置:网站首页 java redis集群 Redis集群下过期key监听

Redis集群下过期key监听

2023-12-12 02:10| 来源: 网络整理| 查看: 265

背景

基于Redis的主动事件的处理,比如:当用户购买了会员卡十分钟内没有付款,需要通过小程序或者APP向用户主动推送购买会员卡的优势,引导用户继续完成支付并购买等,类似的场景需要用户在指定的时间点后主动通知或者继续引导,使用 Redis过期键Event优雅、快捷的实现

实战

在Redis中有两种通知:

类型:

键空间通知:Keyspace@_:mykey(针对mykey的所有的操作的通知)键事件通知:Keyevent@_:expired (键的过期事件)

解释:

KeySpace键空间通知。即针对指定key发生的一切改动,推送给订阅的客户端,侧重于针对指定Key的操作;比如Keyspace@0:mykey,当更新、删除mykey时,会推送该事件keyEvent键时间通知。侧重于指定的事件,如:expired(过期事件)、DEL(删除键事件)等;所有的键不针对指定的键,如果要筛选键要通过代码完成

实现方式共有两种方式:(基于Spring Boot完成)RedisMessageListenerContainer 在Spring Boot 启动之后,最后通过(SmartLifecycle)启动,通过定时任务每隔两秒执行一次订阅的任务,一旦有消息,则遍历所有的MessageListener,通过OnMessage方法执行;所以:

继承MessageListenerAdapter 监听器,重写OnMessage方法继承KeyExpirationEventMessageListener监听器, 重写doHandleMessage方法 以下分别详细讲述实现方式:(关于Redis相关数据源的配置,暂不包括)

MessageListenerAdapter监听器

通过适配器自定义监听器后,需要单独配置RedisMessageListerContainer的MessageListener以及监听Pattern(Keyspace或者Keyevent、db、事件),如下:

@Bean @ConditionalOnMissingBean(RedisMessageListenerContainer.class) public RedisMessageListenerContainer container(){ RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer(); //连接池的设置,暂不描述 listenerContainer.setConnectionFactory(jedisConnectionFactory()); //添加自定义的监听器,并将指定的事件添加进去 listenerContainer.addMessageListener(new MyKeyExpireListener(),new PatternTopic("__keyevent@12__:expired")); return listenerContainer; }

KeyExpirationEventMessageListener监听器

KeyExpirationEventMessageListener继承自KeyspaceEventMessageListener,而KeyspaceEventMessageListener监听所有db的Keyevent,而KeyExpirationEventMessageListener专门监听Key的过期事件的类;然后基于Spring Boot 的事件通知机制,当监听到所有db库中的key有过期事件时,通过ApplicationContext发送RedisKeyExpiredEvent事件,消息体即Key;通过继承KeyExpirationEventMessageListener无需再设置RedisMEssageListenerContainer,通过@Compont注解,让IOC容器发现即可

缺点:

Key的过期事件通知,无法将value也通知到,消息体只有key,在某些场景下,需要将key-Value在储存另外一份作为映射,或者拿到key以后再基于数据库进行操作;

 

1. 前言

在使用redis集群时,发现过期key始终监听不到。网上也没有现成的解决方案。于是想,既然不能监听集群,那我可以建立多个redis连接,分别对每个redis的key过期进行监听。以上做法可能不尽人意,目前也没找到好的解决方案,如果有好的想法,请留言告知哦!不多说,直接贴我自己的代码!

2. 代码实现

关于Redis集群配置代码此处不贴,直接贴配置监听类代码!

redis.host1: 10.113.56.68 redis.port1: 7030 redis.host2: 10.113.56.68 redis.port2: 7031 redis.host3: 10.113.56.68 redis.port3: 7032 redis.host4: 10.113.56.68 redis.port4: 7033 redis.host5: 10.113.56.68 redis.port5: 7034 redis.host6: 10.113.56.68 redis.port6: 7035 application配置类 import org.springframework.beans.factory.annotation.Value; import org.springframework.cache.CacheManager; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.cache.RedisCacheManager; import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.serializer.StringRedisSerializer; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPoolConfig; import java.util.Arrays; /** * @Author xiabing5 * @Create 2019/8/6 14:46 * @Desc 监听redis中Key过期事件 **/ @Configuration public class RedisListenerConfig { @Value("${redis.host1}") private String host1; @Value("${redis.host2}") private String host2; @Value("${redis.host3}") private String host3; @Value("${redis.host4}") private String host4; @Value("${redis.host5}") private String host5; @Value("${redis.host6}") private String host6; @Value("${redis.port1}") private int port1; @Value("${redis.port2}") private int port2; @Value("${redis.port3}") private int port3; @Value("${redis.port4}") private int port4; @Value("${redis.port5}") private int port5; @Value("${redis.port6}") private int port6; @Bean JedisPoolConfig jedisPoolConfig(){ JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); jedisPoolConfig.setMaxIdle(100); jedisPoolConfig.setMaxWaitMillis(1000); return jedisPoolConfig; } // redis-cluster不支持key过期监听,建立多个连接,对每个redis节点进行监听 @Bean RedisMessageListenerContainer redisContainer1() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host1); jedisConnectionFactory.setPort(port1); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisMessageListenerContainer redisContainer2() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host2); jedisConnectionFactory.setPort(port2); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisMessageListenerContainer redisContainer3() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host3); jedisConnectionFactory.setPort(port3); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisMessageListenerContainer redisContainer4() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host4); jedisConnectionFactory.setPort(port4); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisMessageListenerContainer redisContainer5() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host5); jedisConnectionFactory.setPort(port5); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisMessageListenerContainer redisContainer6() { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host6); jedisConnectionFactory.setPort(port6); jedisConnectionFactory.setPoolConfig(jedisPoolConfig()); jedisConnectionFactory.afterPropertiesSet(); container.setConnectionFactory(jedisConnectionFactory); return container; } @Bean RedisKeyExpirationListener redisKeyExpirationListener1() { return new RedisKeyExpirationListener(redisContainer1()); } @Bean RedisKeyExpirationListener redisKeyExpirationListener2() { return new RedisKeyExpirationListener(redisContainer2()); } @Bean RedisKeyExpirationListener redisKeyExpirationListener3() { return new RedisKeyExpirationListener(redisContainer3()); } @Bean RedisKeyExpirationListener redisKeyExpirationListener4() { return new RedisKeyExpirationListener(redisContainer4()); } @Bean RedisKeyExpirationListener redisKeyExpirationListener5() { return new RedisKeyExpirationListener(redisContainer5()); } @Bean RedisKeyExpirationListener redisKeyExpirationListener6() { return new RedisKeyExpirationListener(redisContainer6()); } } Bean配置类 import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.listener.KeyExpirationEventMessageListener; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import java.util.Date; /** * @Author xiabing5 * @Create 2019/9/4 9:47 * @Desc redis过期监听 **/ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener { @Autowired RedisUtil redisUtil; @Autowired LoginUserStatisticsMapper loginUserStatisticsMapper; public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) { super(listenerContainer); } @Override public void onMessage(Message message, byte[] pattern) { // 用户做自己的业务处理即可,message.toString()可以获取失效的key String mesg = message.toString(); } } 监听操作类 3. Redis防止过期key重复监听

对于项目集群情况下,部署多个服务后,容易出现redis过期被多个服务同时监听到,从而执行相同的业务逻辑,这不是我们期望的。单机部署下方法的同步可以采用synchronize关键字。但集群下,就得采用分布式锁。在需要加锁的地方,只要加锁和解锁即可。此处正好写到Redis,那就贴一个自己用的redis分布式锁。

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import redis.clients.jedis.Jedis; import java.util.Collections; import java.util.UUID; /** * @Author xiabing5 * @Create 2019/9/6 15:54 * @Desc redis分布式锁 **/ @Component public class RedisLock { @Autowired Jedis jedis; private static final String SET_IF_NOT_EXIST = "NX"; // NX表示如果不存在key就设置value private static final String SET_WITH_EXPIRE_TIME = "PX"; // PX表示毫秒 // 加锁 public String tryLock(String key,Long acquireTimeout) { // 生成随机value String identifierValue = UUID.randomUUID().toString(); // 设置超时时间 Long endTime = System.currentTimeMillis() + acquireTimeout; // 循环获取锁 while (System.currentTimeMillis() < endTime) { String result = jedis.set(key,identifierValue, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, acquireTimeout); if("OK".equals(result)) { return identifierValue; } } return null; } // 解锁 // public void delLock(String key,String identifierValue) { // // 判断是否是同一把锁 // try{ // if(jedis.get(key).equals(identifierValue)){ // // 此处操作非原子性,容易造成释放非自己的锁 // jedis.del(key); // } // }catch(Exception e) { // e.printStackTrace(); // } // } // 使用Lua代码解锁 public void delLock(String key,String identifierValue) { try{ String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; Long result = (Long) jedis.eval(script, Collections.singletonList(key), Collections.singletonList(identifierValue)); if (1 == result) { System.out.println(result+"释放锁成功"); } if (0 == result) { System.out.println(result+"释放锁失败"); } }catch (Exception e) { e.printStackTrace(); } } } Redis锁解决分布式同步问题

 

4. 总结

自己实现的一个小demo,废话比较少。小白自己写的配置类,理解有问题请留言!自己实现的方案感觉不妥,只是基本完成需求,还得继续研究。



【本文地址】


今日新闻


推荐新闻


CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3